package org.example.checkpoint;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.example.flink.pojo.WaterSensor;
import org.example.function.WaterSensorFunction;

import java.time.Duration;
import java.util.Objects;

/**
 * @Author: tang
 * @Description:
 * @Date 2025/2/26 11:07
 */
public class CheckpointConfigDemo01 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment executionEnvironment =
                StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());

        // 开启 对齐检查点
        // 1、启用检查点: 默认是barrier对齐的，周期为5s, 精准一次
        executionEnvironment.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        CheckpointConfig checkpointConfig = executionEnvironment.getCheckpointConfig();
        // 2、指定检查点的存储位置
        checkpointConfig.setCheckpointStorage("file:///usr/local/flink/data");
        // 3、checkpoint的超时时间: 默认10分钟
        checkpointConfig.setCheckpointTimeout(10000);
        // 4、同时运行中的checkpoint的最大数量
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        // 7、允许 checkpoint 连续失败的次数，默认0--》表示checkpoint一失败，job就挂掉
        checkpointConfig.setTolerableCheckpointFailureNumber(10);
        // 5、最小等待间隔: 上一轮checkpoint结束 到 下一轮checkpoint开始 之间的间隔，设置了>0,并发就会变成1
        checkpointConfig.setMinPauseBetweenCheckpoints(5000);
        // 6、取消作业时，checkpoint的数据 是否保留在外部系统
        // DELETE_ON_CANCELLATION:主动cancel时，删除存在外部系统的chk-xx目录 （如果是程序突然挂掉，不会删）
        // RETAIN_ON_CANCELLATION:主动cancel时，外部系统的chk-xx目录会保存下来
        checkpointConfig.setExternalizedCheckpointCleanup(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        // 非对齐检查点
        executionEnvironment.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        checkpointConfig.enableUnalignedCheckpoints();
        checkpointConfig.setAlignedCheckpointTimeout(Duration.ofSeconds(5));


        SingleOutputStreamOperator<String> streamOperator = executionEnvironment
                .socketTextStream("127.0.0.1", 7777)
                .name("socket").uid("uid-socket")
                .map(new WaterSensorFunction())
                .name("map").uid("uid-map")
                .keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {

                    ValueState<Integer> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        valueState = getRuntimeContext()
                                .getState(new ValueStateDescriptor<Integer>("valueState", Types.INT));
                    }

                    @Override
                    public void processElement(WaterSensor waterSensor,
                                               KeyedProcessFunction<String, WaterSensor, String>.Context ctx,
                                               Collector<String> out) throws Exception {
                        Integer value = valueState.value();
                        valueState.update((Objects.isNull(value) ? 0 : value) + 1);
                        out.collect("当前id: " + ctx.getCurrentKey() + ",sum=" + valueState.value());
                    }
                }).name("process").uid("uid-process");

        streamOperator.print().name("print").uid("uid-print");

        executionEnvironment.execute();

    }

}
